package com.atguigu.gmall.realtime.dws.app;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.base.BaseApp;
import com.atguigu.gmall.realtime.common.bean.TrafficHomeDetailPageViewBean;
import com.atguigu.gmall.realtime.common.constant.Constant;
import com.atguigu.gmall.realtime.common.function.BeanToJsonMapFunction;
import com.atguigu.gmall.realtime.common.util.DateFormatUtil;
import com.atguigu.gmall.realtime.common.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * 从 Kafka页面日志主题读取数据，统计当日的首页和商品详情页独立访客数。
 */
public class DwsTrafficHomeDetailPageViewWindow extends BaseApp {
    public static void main(String[] args) {
        new DwsTrafficHomeDetailPageViewWindow().start(10023, 4, "dws_traffic_home_detail_page_view_window", Constant.TOPIC_DWD_TRAFFIC_PAGE);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {
        // 读取数据

        // 转换数据结构, 将流中的数据转换为JSONObject
        SingleOutputStreamOperator<JSONObject> map = stream.map(
                new MapFunction<String, JSONObject>() {
                    @Override
                    public JSONObject map(String s) throws Exception {
                        return JSON.parseObject(s);
                    }
                }
        );
        // map.print("MAP");


        // 过滤数据, 仅保留 page_id 为 home 或 good_detail 的数据  因为本程序统计的度量仅与这两个页面有关，其它数据无用。
        // 按照 mid 分组
        // 统计首页和商品详情页独立访客数，转换数据结构
        SingleOutputStreamOperator<TrafficHomeDetailPageViewBean> processDs = map.keyBy(
                new KeySelector<JSONObject, String>() {
                    @Override
                    public String getKey(JSONObject jsonObject) throws Exception {
                        return jsonObject.getJSONObject("common").getString("mid");
                    }
                }
        ).process(
                new KeyedProcessFunction<String, JSONObject, TrafficHomeDetailPageViewBean>() {
                    private ValueState<String> homeState;
                    private ValueState<String> goodDetailState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        RuntimeContext runtimeContext = getRuntimeContext();
                        ValueStateDescriptor<String> homeStateDesc = new ValueStateDescriptor<>("homeStateDesc", Types.STRING);
                        ValueStateDescriptor<String> goodDetailStateDesc = new ValueStateDescriptor<>("goodDetailStateDesc", Types.STRING);
                        homeState = runtimeContext.getState(homeStateDesc);
                        goodDetailState = runtimeContext.getState(goodDetailStateDesc);
                    }

                    @Override
                    public void processElement(JSONObject jsonObject, KeyedProcessFunction<String, JSONObject, TrafficHomeDetailPageViewBean>.Context context, Collector<TrafficHomeDetailPageViewBean> collector) throws Exception {
                        /**
                         * 如果 page_id 为 home，当状态中存储的日期为 null 或不是当日时，
                         * 将 homeUvCt（首页独立访客数） 置为 1，并将状态中的日期更新为当日。
                         * 否则置为 0，不做操作。商品详情页独立访客的统计同理。
                         * 当 homeUvCt 和 detailUvCt 至少有一个不为 0 时，
                         * 将统计结果和相关维度信息封装到定义的实体类中，发送到下游，否则舍弃数据。
                         */

                        // 保留 page_id 为 home 或 good_detail 的数据
                        // 获取当前数据的page_id
                        String pageId = jsonObject.getJSONObject("page").getString("page_id");

                        Long ts = jsonObject.getLong("ts");

                        String today = DateFormatUtil.tsToDate(ts);

                        String lastHomeDate = homeState.value();
                        String lastGoodDetailDate = goodDetailState.value();

                        Long homeUvCt = 0L;
                        Long goodDetailUvCt = 0L;

                        if (pageId.equals("home") && !today.equals(lastHomeDate)) {  // 是首页,且和今天状态不相等,即是第一条数据
                            homeUvCt = 1L;
                            homeState.update(today);
                        } else if (pageId.equals("good_detail") && !today.equals(lastGoodDetailDate)) {
                            goodDetailUvCt = 1L;
                            goodDetailState.update(today);
                        }

                        // 如果都为零, 数据舍弃, 反之收集
                        if (homeUvCt + goodDetailUvCt == 1) {
                            TrafficHomeDetailPageViewBean trafficHomeDetailPageViewBean = TrafficHomeDetailPageViewBean.builder()
                                    .stt("")
                                    .edt("")
                                    .curDate("")
                                    .homeUvCt(homeUvCt)
                                    .goodDetailUvCt(goodDetailUvCt)
                                    .ts(ts)
                                    .build();
                            collector.collect(trafficHomeDetailPageViewBean);
                        }

                    }
                }
        );
        // processDs.print("PRICESSDS");

        // 设置水位线
        SingleOutputStreamOperator<TrafficHomeDetailPageViewBean> trafficHomeDetailPageViewBeanSingleOutputStreamOperator = processDs.assignTimestampsAndWatermarks(
                WatermarkStrategy.<TrafficHomeDetailPageViewBean>forBoundedOutOfOrderness(Duration.ofSeconds(5L))
                        .withTimestampAssigner((bean, ts) -> bean.getTs())
                        .withIdleness(Duration.ofSeconds(120L))
        );
        // trafficHomeDetailPageViewBeanSingleOutputStreamOperator.print();


        // 开窗 按键 滚动时间
        SingleOutputStreamOperator<TrafficHomeDetailPageViewBean> reduceDs = trafficHomeDetailPageViewBeanSingleOutputStreamOperator.windowAll(
                TumblingEventTimeWindows.of(Time.seconds(5L))
        ).reduce(
                new ReduceFunction<TrafficHomeDetailPageViewBean>() {
                    @Override
                    public TrafficHomeDetailPageViewBean reduce(TrafficHomeDetailPageViewBean t1, TrafficHomeDetailPageViewBean t2) throws Exception {
                        t1.setHomeUvCt(t1.getHomeUvCt() + t2.getHomeUvCt());
                        t1.setGoodDetailUvCt(t1.getGoodDetailUvCt() + t2.getGoodDetailUvCt());
                        return t1;
                    }
                },
                // 聚合
                new ProcessAllWindowFunction<TrafficHomeDetailPageViewBean, TrafficHomeDetailPageViewBean, TimeWindow>() {
                    @Override
                    public void process(ProcessAllWindowFunction<TrafficHomeDetailPageViewBean, TrafficHomeDetailPageViewBean, TimeWindow>.Context context, Iterable<TrafficHomeDetailPageViewBean> iterable, Collector<TrafficHomeDetailPageViewBean> collector) throws Exception {
                        // 获取窗口信息
                        TrafficHomeDetailPageViewBean bean = iterable.iterator().next();
                        bean.setStt(DateFormatUtil.tsToDateTime(context.window().getStart()));
                        bean.setEdt(DateFormatUtil.tsToDateTime(context.window().getEnd()));
                        bean.setCurDate(DateFormatUtil.tsToDate(context.window().getStart()));
                        collector.collect(bean);
                    }
                }
        );
        // reduceDs.print("reduceDs");


        // 写出数据到doris
        // 先转换格式为Json再写入
        SingleOutputStreamOperator map1 = reduceDs.map(
                new BeanToJsonMapFunction<>()
        );
        map1.print("MAP!");
        map1.sinkTo(FlinkSinkUtil.getDorisSink(Constant.DORIS_DATABASE, Constant.DWS_TRAFFIC_HOME_DETAIL_PAGE_VIEW_WINDOW));


    }
}
